package goroutine

import (
	"context"
	"runtime"
	"sync"
	"sync/atomic"
	"time"

	"gitee.com/doraemon1996/bunny/log"
	"gitee.com/doraemon1996/bunny/structure"
)

// 工作协程过期清理时间
const ExpireIntervalTime = time.Second

var (
	// 协程池最大容量
	MaxGoroutine int32 = 1000000

	// 工作协程任务通道容量
	workerTaskCap = func() int {
		if runtime.GOMAXPROCS(0) == 1 {
			return 0
		}
		return 1
	}()
)

// 协程池
type GoPool struct {
	ctl *structure.Control

	// 协程池最大容量
	capacity int32

	// 运行中协程数量
	running int32

	// 等待中任务数量
	waiting int32

	// 条件变量锁
	cond *sync.Cond

	// 空闲协程管理器
	idleMng idleMng

	// 工作协程对象池
	workerPool sync.Pool
}

// 创建一个协程池
func New(ctx context.Context, size int) *GoPool {
	pool := &GoPool{
		ctl:      new(structure.Control),
		capacity: MaxGoroutine,
		running:  0,
		waiting:  0,
		cond:     sync.NewCond(&sync.Mutex{}),
		idleMng:  newIdleMng(size),
	}

	pool.workerPool.New = func() any {
		return &worker{
			goPool: pool,
			pipe:   make(chan func(), workerTaskCap),
		}
	}

	pool.ctl.Init(ctx, nil)
	pool.ctl.Start(func() {
		pool.ctl.Add(1)
		go func() {
			defer pool.ctl.Done()

			pool.tick()
		}()
	})

	return pool
}

// 获取运行中的协程数量
func (p *GoPool) GetRunning() int32 {
	return atomic.LoadInt32(&p.running)
}

// 获取等待任务数量
func (p *GoPool) GetWaiting() int32 {
	return atomic.LoadInt32(&p.waiting)
}

// 获取空闲协程数量
func (p *GoPool) GetIdling() int32 {
	return p.capacity - p.GetRunning()
}

// 空闲协程清理循环
func (p *GoPool) tick() {
	ticker := time.NewTicker(ExpireIntervalTime)
	defer ticker.Stop()

	for !p.ctl.Closed() {
		<-ticker.C

		p.cond.L.Lock()
		expireList := p.idleMng.Expire(ExpireIntervalTime)
		p.cond.L.Unlock()

		for i, worker := range expireList {
			expireList[i] = nil
			worker.pipe <- nil
		}

		if p.GetWaiting() > 0 && p.GetIdling() > 0 {
			p.cond.Broadcast()
		}
	}
}

// 关闭并清空协程池
func (p *GoPool) Stop() {
	p.ctl.Stop(func() {
		p.cond.L.Lock()
		list := p.idleMng.Clear()
		p.cond.L.Unlock()

		// 立即唤醒等待提交任务的地方,使自己退出
		// 理论上来说,关闭协程池时,不应该还有任务等待执行
		p.cond.Broadcast()

		for _, worker := range list {
			worker.pipe <- nil
		}

		p.ctl.Wait()
	})
}

// 增加运行中工作协程数量
func (p *GoPool) addRunning(num int32) {
	atomic.AddInt32(&p.running, num)
}

// 增加等待任务数量
func (p *GoPool) addWaiting(num int32) {
	atomic.AddInt32(&p.waiting, num)
}

// 从缓存池种获取工作协程对象
func (p *GoPool) spanWorker() *worker {
	worker := p.workerPool.Get().(*worker)
	worker.run()
	return worker
}

// 从获取一个工作协程
func (p *GoPool) getWorker() (worker *worker) {
	p.cond.L.Lock()
	defer p.cond.L.Unlock()

	worker = p.idleMng.GetWorker()
	if worker == nil {
		if p.capacity > p.GetRunning() {
			// 协程池没有满,可以继续分配
			worker = p.spanWorker()
		} else {
			// 协程池满了,等待工作协程空闲
			for {
				p.addWaiting(1)
				p.cond.Wait() // 阻塞等待可用工作线程
				p.addWaiting(-1)

				worker = p.idleMng.GetWorker()
				if worker != nil {
					break
				} else if p.capacity > p.GetRunning() {
					// 协程池没有满,可以继续分配
					worker = p.spanWorker()
					break
				}
			}
		}
	}

	return
}

// 将一个工作协程放进空闲协程管理器
func (p *GoPool) putWorker(worker *worker) bool {
	p.cond.L.Lock()
	defer p.cond.L.Unlock()

	// 协程池已关闭 或 达到协程池容量上限
	if p.ctl.Closed() || p.capacity <= p.GetRunning() {
		p.cond.Broadcast()
		return true
	}

	// 空闲管理器已满
	if p.idleMng.IsFull() {
		p.cond.Broadcast()
		return true
	}

	worker.idleTime = time.Now()
	p.idleMng.PutWorker(worker)

	p.cond.Signal()

	return false
}

// 提交任务
func (p *GoPool) Submit(task func()) {
	if p.ctl.Closed() {
		log.Error("call Submit() with closed GoPool, will excute task function direct")
		task()
		return
	}

	worker := p.getWorker()
	worker.pipe <- task
}
